package hook

import (
	"errors"
	"jtdevice/app/db"
	"jtdevice/pkg/mqt"
	"strings"

	"github.com/bytedance/sonic"
)

func onMsgOnline(b []byte) error {
	var s mqMsgOnline
	if err := sonic.Unmarshal(b, &s); err != nil {
		return err
	}
	db.Dispatcher().PublishOnline(s.devModelOnline())
	return nil
}

func onMsgStatus(b []byte) error {
	var s mqMsgStatus
	if err := sonic.Unmarshal(b, &s); err != nil {
		return err
	}
	db.Dispatcher().PublishStatus(s.devModelStatus())
	return nil
}

func onMsgAlarm(b []byte) error {
	var s mqMsgAlarm
	if err := sonic.Unmarshal(b, &s); err != nil {
		return err
	}
	db.Dispatcher().PublishAlarm(s.devModelAlarm())
	return nil
}

type receiver struct {
	mqcli *mqt.Client // TODO多mq推送
}

var Receiver receiver

func (o *receiver) Run(subscribe bool) (err error) {
	if !subscribe {
		return nil
	}
	if strings.Contains(opt.Address, "nats://") {
		o.mqcli, err = mqt.News("nats", &mqt.Options{Address: opt.Address}, 3)
	}
	if o.mqcli == nil {
		return errors.New("unsupported MQ")
	}
	// TODO这里消费性能
	o.mqcli.Subscribe(opt.Status, onMsgStatus)
	o.mqcli.Subscribe(opt.Alarm, onMsgAlarm)
	o.mqcli.Subscribe(opt.Online, onMsgOnline)
	return
}

// 如果已开启订阅，不直接入库
func (o *receiver) handleMsg(topic string, b []byte) {
	if !db.Dispatcher().IsRunning() {
		return
	}
	if o.mqcli != nil {
		return
	}
	switch topic {
	case opt.Alarm:
		onMsgAlarm(b)
	case opt.Online:
		onMsgOnline(b)
	case opt.Status:
		onMsgStatus(b)
	}
}

func (o *receiver) Release() {
	if o.mqcli != nil {
		o.mqcli.Release()
	}
}
